use std::sync::Arc;
use std::time::SystemTime;
use linked_hash_map::LinkedHashMap;
use log::{error};
use tokio::io::{AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::sync::{broadcast, mpsc, oneshot};
use rpc::connection_manager::connection::RpcConnection;
use rpc::router::{RouterFactory};
use rpc::{time_from_millis, time_to_millis, ChannelId, DefaultLocalSeqCreator, LocalSeqCreator, RpcContext, RpcEvent, RpcFilter, RpcResult};
use rpc::bootstrap::Bootstrap;
use rpc::dto::{read_packet_for_timeout, str_to_packet};
use rpc::error::{CustomerErrorKind, RpcError};
use rpc::error::RpcError::CustomerError;
use rpc::sender::RpcSender;
use crate::connection_manager::client_connection::{ClientConnectionHandler, RpcClientChannel};
use crate::connection_manager::ClientConnectionManager;

pub struct SocketClient<'a> {
    address: &'a str,
    current_codec_type: &'static str,
    bootstrap: Bootstrap,
    shutdown: broadcast::Sender<()>,
}

impl <'a> Drop for SocketClient<'a> {
    fn drop(&mut self) {
        println!("Dropping socket client");
    }
}

impl <'a> SocketClient<'a> {

    pub fn get_router(&self, path: &'static str) ->  Option<&Box<dyn RouterFactory + 'static>> {
        self.bootstrap.get_router(path)
    }

    pub fn get_filters(&self)-> Arc<LinkedHashMap<&'static str, Box<dyn RpcFilter + 'static>>> {
        self.bootstrap.get_filters()
    }

    pub async fn connection(&self) -> Result<RpcClientChannel, RpcError> {
        let mut stream = TcpStream::connect(self.address).await?;
        let client_connection_manager = ClientConnectionManager::new();
        let client_manager_sender = rpc::connection_manager::start(client_connection_manager);

        let (event_sender, event_receive) = mpsc::unbounded_channel::<Box<dyn RpcEvent>>();
        let sender = RpcSender::new(client_manager_sender.clone(), event_sender);

        let router_clone = self.bootstrap.router.clone();
        let buff_size = self.bootstrap.buff_size;
        let rpc_codec_map = self.bootstrap.rpc_codec_map.clone();

        let sender_clone = sender.clone();

        let (oneshot_sender, oneshot_receiver) = oneshot::channel::<bool>();

        let server_time = self.write_codec_type_data(&mut stream).await?;

        let creator = match self.bootstrap.local_seq_creator.clone() {
            None => {
                let prefix= format!("{}", time_to_millis(server_time));
                let t = Box::new(DefaultLocalSeqCreator::new(prefix)) as Box<dyn LocalSeqCreator>;
                Arc::new(t)
            }
            Some(local_seq_creator) => {local_seq_creator }
        };

        let mut context = RpcContext::create(creator, rpc_codec_map, sender_clone.clone());
        context.set_current_codec_type(Some(self.current_codec_type.to_string())).unwrap();
        let context_clone = context.clone();

        let idle_listeners = self.bootstrap.idle_listeners.clone();

        let (send, receive) = mpsc::unbounded_channel::<()>();

        let close_receive = self.shutdown.subscribe();
        tokio::spawn(async move {
            let mut connection = RpcConnection::new(ChannelId::new(0), buff_size,
                    context_clone.clone(), event_receive, idle_listeners);
            let connection_handler = ClientConnectionHandler::new(router_clone, context_clone);

            let result = connection.run(stream, connection_handler,
                                        || { oneshot_sender.send(true).unwrap();}, close_receive).await;
            if let Err(e) = result {
                error!("connection error: {}", e);
            }
        });

        let is_open_success = oneshot_receiver.await;
        match is_open_success {
            Ok(_is_open_success) => {}
            Err(error) => { return Err(error.into());}
        }

        let rpc_channel = RpcClientChannel::new(context);

        Ok(rpc_channel)
    }

    pub async fn close(&self) -> Result<(), RpcError> {
        self.shutdown.send(()).unwrap();
        Ok(())
    }

    async fn write_codec_type_data(&self, stream: &mut TcpStream) -> RpcResult<SystemTime> {
        let write_data = str_to_packet(self.current_codec_type);
        stream.write_all(&write_data).await?;

        let packet = read_packet_for_timeout(stream, 3000usize).await?;
        let value = String::from_utf8(packet).unwrap();
        if !value.starts_with("success:") {
            let error = CustomerError {error_kid: CustomerErrorKind::CodecTypeNoSetError,
                message: "编码设置错误".to_string()};
            return Err(error);
        }
        let (_, time_u64) = value.split_at("success:".len());
        let time = time_u64.parse::<u64>()?;
        Ok(time_from_millis(time))
    }
}

pub struct RpcClientBuilder<'a> {
    address: &'a str,
    current_codec_type: Option<&'static str>,
    bootstrap: Bootstrap,
}

impl <'a> RpcClientBuilder<'a> {

    pub fn new(bootstrap: Bootstrap, address: &'a str) -> Self {
        let builder = RpcClientBuilder {
            bootstrap,
            address,
            current_codec_type: None,
        };
        builder
    }

    pub fn set_current_codec_type(self, codec_type: &'static str) -> Self {
        Self {
            current_codec_type: Some(codec_type),
            ..self
        }
    }

    pub fn build(self) -> SocketClient<'a> {
        if let None = self.current_codec_type {
            panic!("请设置编码类型")
        }

        let (close_send, close_receive) = broadcast::channel::<()>(1);
        SocketClient {
            address: self.address,
            current_codec_type: self.current_codec_type.unwrap(),
            bootstrap: self.bootstrap,
            shutdown: close_send,
        }
    }
}